package com.foreveross.fetl.queue;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.log4j.Logger;
import org.springframework.stereotype.Component;

import com.foreveross.fetl.common.HBaseTable;
import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.common.vo.AirLineCountVO;
import com.foreveross.fetl.common.vo.FetlErrorMessage;
import com.foreveross.fetl.core.domain.task.SourceEntity;
import com.foreveross.fetl.queue.util.LogMessageUtils;
import com.foreveross.fetl.util.AirLineCountUtils;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.FetlErrorUtils;
import com.foreveross.fetl.util.RowkeyUtils;
import com.foreveross.fetl.util.StringUtils;

/**
 * "99999yyyyMMdd999999" 表示yyyyMMdd 天抓到的总数
 * "20005yyyyMMdd999999" 表示 渠道20005在yyyyMMdd天抓到的总数
 * 
 * @author ljshi
 *
 */
@Component
public class AirLineCountProcess {
	
	private static Logger log = Logger.getLogger("AirLineCountProcess");
	
	
	private int tasksNumber;//任务数 多少条rowkey
	
	private int failuresNumber;//失败数
	
	private int insertNumber; //插入条数 
	
	private long excuteStartTime;
	
	
	
	public void  rollBackNumber(){
		tasksNumber=0;
		failuresNumber=0;
		insertNumber=0;
		excuteStartTime = System.currentTimeMillis();
		LogMessageUtils.create("航线抓取情况统计", this.getClass().getSimpleName());
	}
	
	public void manualExecute(String date){
		try {
			this.rollBackNumber();
			if(null==date)
				return;
			Date starDate=DateUtils.parse(date.replace("-", ""), "yyyyMMdd");
			Calendar c=Calendar.getInstance();
			c.setTime(starDate);
			c.add(Calendar.DAY_OF_MONTH, 1);
			this.process(starDate,c.getTime());
			
		} catch (Exception e) {
			LogMessageUtils.build("航线抓取情况统计", this.getClass().getSimpleName(), 3, StringUtils.getStackTrace(e), failuresNumber, tasksNumber, 
					insertNumber, date, excuteStartTime);
			
			//错误日志  保存到Hbase里
			FetlErrorMessage vo=FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.airLineCount, 
					this.getClass().getSimpleName(),DateUtils.format(new Date(), "yyyyMMddHH"), e);
			FetlErrorUtils.saveToHbase(vo);
		}
		LogMessageUtils.build("航线抓取情况统计", this.getClass().getSimpleName(), 2, null, failuresNumber, tasksNumber, insertNumber, date, excuteStartTime);
	}
	
	public void execute(){
		rollBackNumber();
		Date starDate=null;
		try {
//			starDate = this.getLastDateforHbase(HBaseTable.TABLE_AIRLINE_COUNT);
//			if(null==starDate)
//				starDate=DateUtils.parse("20130401", "yyyyMMdd");
			Calendar c=Calendar.getInstance();
			c.add(Calendar.DAY_OF_MONTH, -1);
			starDate=c.getTime();
			Date today=DateUtils.parse(DateUtils.format(new Date(), "yyyyMMdd"), "yyyyMMdd");
			this.process(starDate,today);
		} catch (Exception e) {
			log.error(StringUtils.getStackTrace(e));
			//处理日志  保存到mysql 中 
			LogMessageUtils.build("航线抓取情况统计", this.getClass().getName(), 3, StringUtils.getStackTrace(e), failuresNumber, tasksNumber, 
					insertNumber, DateUtils.format(starDate, "yyyy-MM-dd"), excuteStartTime);
			
			//错误日志  保存到Hbase里
			FetlErrorMessage vo=FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.airLineCount, this.getClass().getSimpleName(),DateUtils.format(new Date(), "yyyyMMddHH"), e);
			FetlErrorUtils.saveToHbase(vo);
		}
		LogMessageUtils.build("航线抓取情况统计", this.getClass().getSimpleName(), 2, null, failuresNumber, tasksNumber, insertNumber, DateUtils.format(starDate, "yyyy-MM-dd"), excuteStartTime);
	}
	
	/**
	 * 处理时间：从lastDate的0点开始到今天的0点
	 * @param starDate
	 */
	private void process(Date starDate ,Date endDate) {
		SourceEntity source = new SourceEntity();
		List<SourceEntity> listSource=source.findAll();
		Calendar cStart = Calendar.getInstance();
		cStart.setTime(starDate);
		Calendar cEnd = Calendar.getInstance();
		cEnd.setTime(endDate);
		while (cStart.before(cEnd)) {
			for(SourceEntity sourceEntity :listSource){
				this.process(cStart.getTime(),sourceEntity,true);//单程
				this.process(cStart.getTime(),sourceEntity,false);//往返
			}
			this.countAll(cStart.getTime(),listSource,true);//单程
			this.countAll(cStart.getTime(),listSource,false);//往返
			cStart.add(Calendar.DAY_OF_MONTH, 1);//加一天
		}
		
	}
	
	/**
	 * 汇总starDate的统计结果
	 * "20005yyyyMMdd999999" 表示 渠道20005在yyyyMMdd天抓到的总数
	 * @param starDate
	 * @param listSource 
	 */
	private void countAll(Date starDate, List<SourceEntity> listSource,boolean isOneWay) {
		
		String tableName=HBaseTable.TABLE_AIRLINE_COUNT;
		if(!isOneWay)
			tableName=HBaseTable.TABLE_AIRLINE_COUNT_DOUBLE;
		int count=0;
		String dateStr=DateUtils.format(starDate, "yyyyMMdd");
		Result res=null;
		for(SourceEntity source:listSource){
			String rowkey=source.getHbaseKey()+dateStr+"999"+"999";
			try {
				res=HBaseHelper.getOneRecord(tableName, rowkey);
				Map<String, Map<String, String>> curMap = HBaseHelper
						.transformHBaseRowToMap(res);
				AirLineCountVO vo=AirLineCountUtils.getFormHbase(curMap);
				if(null==vo || vo.getFlightSize()==null)
					continue;
				count+=vo.getFlightSize();
			} catch (IOException e) {
				log.error(StringUtils.getStackTrace(e));
				//错误日志
				FetlErrorMessage vo=FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.airLineCount, this.getClass().getSimpleName(),DateUtils.format(new Date(), "yyyyMMddHH"), e);
				List<String> list =new ArrayList<String>();
				list.add(rowkey);
				vo.setKeys(list);
				FetlErrorUtils.saveToHbase(vo);
			}
		}
	  AirLineCountVO all=AirLineCountVO.buildAirLineCount("99999", "all", DateUtils.format(starDate, "yyyy-MM-dd"), "999", "999");
	  all.setFlightSize(count);
	  AirLineCountUtils.saveToHbase(all, tableName);
	}

	/**
	 * 统计
	 * 处理日期 starDate 抓到的渠道sourceEntity的数据
	 * "99999yyyyMMdd999999" 表示yyyyMMdd 天抓到的总数
	 * @param starDate
	 * @param sourceEntity
	 */
	private void process(Date starDate, SourceEntity sourceEntity,boolean isOneWay) {
		String startRow=RowkeyUtils.getRowKey(DateUtils.format(starDate, "yyyyMMdd"), sourceEntity.getHbaseKey(), RowkeyUtils.START_ROW_KEY);
		String endRow=RowkeyUtils.getRowKey(DateUtils.format(starDate, "yyyyMMdd"), sourceEntity.getHbaseKey(), RowkeyUtils.STOP_ROW_KEY);
		Map<String,Integer> map =new HashMap<String ,Integer>();
		ResultScanner rs = null;
		String tableName=HBaseTable.BASE_DATA_FLIGHT;
		if(!isOneWay)
			tableName="CRAWL_PLANE_FREIGHT_DOUBLE";
		try {
			rs=HBaseHelper.getRecords(tableName, startRow, endRow,
					new String[]{"planeinfo:fromCity","planeinfo:toCity"});
			Map<String, String> curMap =null;
			String rowkey=null;
			for (Result res : rs) {
				rowkey=new String(res.getRow());
				tasksNumber++;
				try {
					curMap=HBaseHelper.transformHBaseRowToMap(res).get("planeinfo");
					String line=curMap.get("fromCity")+"-"+curMap.get("toCity");
					if(map.containsKey(line)){
						int count=map.get(line)+1;
						map.put(line, count);
					}else{
						map.put(line, 1);
					}
				} catch (Exception e) {
					FetlErrorMessage vo=FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.airLineCount, this.getClass().getName(),DateUtils.format(new Date(), "yyyyMMddHH"), e);
					List<String> list =new ArrayList<String>();
					list.add(rowkey);
					vo.setKeys(list);
					FetlErrorUtils.saveToHbase(vo);
					failuresNumber++;
					e.printStackTrace();
				}
			}
		 
		} catch (IOException e) {
			log.error(StringUtils.getStackTrace(e));
			FetlErrorMessage vo=FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.airLineCount, this.getClass().getName(),DateUtils.format(new Date(), "yyyyMMddHH"), e);
			vo.setStartRowKey(startRow);
			vo.setEndRowkey(endRow);
			FetlErrorUtils.saveToHbase(vo);
		}finally {
			if (rs != null)
				rs.close();
			rs = null;
		}
		
		this.saveToHbase( starDate,  sourceEntity, isOneWay,map);
	}
	
	/**
	 * 将统计结果保存到Hbase
	 * @param starDate
	 * @param sourceEntity
	 * @param isOneWay
	 * @param map
	 */
	private void saveToHbase(Date starDate, SourceEntity source,
			boolean isOneWay, Map<String, Integer> map) {
		if(null== map)
			return;
		String tableName=HBaseTable.TABLE_AIRLINE_COUNT;
		if(!isOneWay)
			tableName=HBaseTable.TABLE_AIRLINE_COUNT_DOUBLE;
		int count=0;
		for(String line:map.keySet()){
			try {
				String[] citys=line.split("-");
				AirLineCountVO vo=AirLineCountVO.buildAirLineCount(source.getHbaseKey(), source.getName(), DateUtils.format(starDate, "yyyy-MM-dd"), 
						citys[0], citys[1]);
				vo.setFlightSize(map.get(line));
				count+=map.get(line);
				AirLineCountUtils.saveToHbase(vo, tableName);
				insertNumber++;
			} catch (Exception e) {
				failuresNumber++;
				log.error(StringUtils.getStackTrace(e));
				FetlErrorMessage vo=FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.airLineCount, this.getClass().getName(),DateUtils.format(new Date(), "yyyyMMddHH"), e);
				FetlErrorUtils.saveToHbase(vo);
			}
		}
		AirLineCountVO vo=AirLineCountVO.buildAirLineCount(source.getHbaseKey(), source.getName(), DateUtils.format(starDate, "yyyy-MM-dd"), 
				"999", "999");
		vo.setFlightSize(count);
		vo.setTripsSize(map.size());
		AirLineCountUtils.saveToHbase(vo, tableName);
	}

	/**
	 * 获得Hbase 里最后的插入的日期+1天 
	 * HBaseTable.TABLE_AIRLINE_COUNT 单程
	 * HBaseTable.BASE_DATA_FLIGHT_DOUBLE 往返
	 * "99999yyyyMMdd999999" 表示yyyyMMdd 天抓到的总数
	 * @return
	 * @throws IOException
	 */
	private Date getLastDateforHbase(String tableName) throws IOException {
		String startRow="99999"+"20130401"+"999"+"999";
		String stopRow="99999"+DateUtils.format(new Date(), "yyyyMMdd")+"999"+"999";
		ResultScanner rs = null;
		try {
			String rowkey=null;
			rs=HBaseHelper.getRecords(tableName, startRow, stopRow);
			for (Result res : rs) {
				rowkey=new String(res.getRow());
			}
			if(null==rowkey || rowkey.length()<13)
				return null;
			Date lastDate=DateUtils.parse(rowkey.substring(5, 13), "yyyyMMdd");
			Calendar c =Calendar.getInstance();
			c.setTime(lastDate);
			c.add(Calendar.DAY_OF_MONTH, 1);
			return c.getTime();
		} catch (IOException e) {
			log.error(StringUtils.getStackTrace(e));
			FetlErrorMessage vo=FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.airLineCount, this.getClass().getName(),DateUtils.format(new Date(), "yyyyMMddHH"), e);
			vo.setStartRowKey(startRow);
			vo.setEndRowkey(stopRow);
			FetlErrorUtils.saveToHbase(vo);
			throw e;
		}finally{
			if(rs != null) rs.close();
			rs = null;
		}
	}
	
}
